package com.lhl.mq.mqServer;

import com.lhl.mq.common.Consumer;
import com.lhl.mq.common.MqException;
import com.lhl.mq.mqServer.core.*;
import com.lhl.mq.mqServer.datacenter.DataBaseManager;
import com.lhl.mq.mqServer.datacenter.DiskDataCenter;
import com.lhl.mq.mqServer.datacenter.MemoryDataCenter;
import com.sun.javafx.tk.TKPulseListener;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class VirtualHost {
    private String virtualHostName;
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private Router router = new Router();
    private ConsumerManage consumerManage = new ConsumerManage(this);

    //操作交换机的锁
    private final Object exchangeLocker = new Object();
    //操作队列的锁
    private final Object queueLock = new Object();

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public VirtualHost(String name){
        this.virtualHostName = name;

        //memoryDataCenter 不需要额外的进行初始化操作，对象new出来就可以了
        //dataBaseCenter 需要进行额外的初始化，需要建库建表以及初始数据的设定
        diskDataCenter.init();

        //建硬盘中的数据恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (MqException | IOException | ClassNotFoundException e ) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据数据失败，virtualHostName = " + virtualHostName);
        }
    }

    //创建交换机
    //如果已经已有一样的交换机就直接返会
    //没有则创建
    public boolean exchangeDeclare(String exchangeName, ExchangeType type, boolean durable,
                               boolean autoDelete, Map<String,Object> arguments){
        //修改传入进来的交换机的名字
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                //判断交换机是否存在
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                //如果存在直接返回
                if (existsExchange != null){
                    System.out.println("[VirtualHost] 交换机已存在，无需再次创建，exchangeName = " + exchangeName);
                }

                //创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(type);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);

                //将交换机添加到硬盘(持久化)
                //先判断是否需要持久化
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }

                //将交换机添加到内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 成功创建交换机， exchangeName = " + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 创建交换机失败， exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //删除交换机
    //存在则删除
    //不存在则抛出异常
    public boolean exchangeDelete(String exchangeName){
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeName) {
                //判断交换机是否存在
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange == null){
                    throw new MqException("[VirtualHost] 删除交互机失败，exchangeName = " + exchangeName);
                }

                //删除硬盘上的交换机
                if (existsExchange.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                //删除内存上的交换机
                memoryDataCenter.deleteExchange(exchangeName);

                System.out.println("[VirtualHost] 成功删除交换机，exchangeName = " + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 删除交换机失败，exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //创建队列
    //存在则直接返回
    //不存在则创建
    public boolean queueDeclare(String queueName,boolean durable,boolean exclusive,boolean autoDelete,Map<String, Object> argument){
        queueName = virtualHostName + queueName;
        try {
            //判断队列是否存在
            synchronized (queueLock) {
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                //存在则直接返回
                if (existsQueue != null){
                    System.out.println("[VirtualHost] 队列已存在，无需再次创建，queueName = " + queueName);
                    return true;
                }

                //创建队列
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(argument);

                //判断是否需要持久化
                if (durable){
                    //需要持久化则写入硬盘
                    diskDataCenter.insertQueue(queue);
                }
                //写入内存
                memoryDataCenter.insertQueue(queue);

                System.out.println("[VirtualHost] 成功创建队列，queueName = " + queueName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 创建队列失败，queueName = " + queueName);
            e.printStackTrace();;
            return false;
        }
    }

    //删除队列
    //判断队列是否存在
    //不存在则抛出异常 存在则进行删除
    public boolean queueDelete(String queueName){
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLock) {
                //判断队列是否存在
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue == null){
                    //不存在，抛异常
                    throw new MqException("[VirtualHost] 删除队列失败，queueName = " + queueName);
                }

                //删除硬盘上的队列
                if (existsQueue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                //删除内存上的队列
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualQueue] 成功删除队列，queueName = " + queueName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 队列删除失败，queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }


    //创建消息绑定
    public boolean queueBind(String queueName,String exchangeName,String bindingKey){
        queueName = virtualHostName + queueName;
        exchangeName =  virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                synchronized (queueLock) {
                    //判断队列和交换机是否存在
                    MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                    Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                    if (existsQueue == null ){
                        throw new MqException("[VirtualHost] 创建绑定失败，队列不存在，queueName = " + queueName);
                    }
                    if (existsExchange == null ){
                        throw new MqException("[VirtualHost] 创建绑定失败，交换机不存在，exchangeName = " + exchangeName);
                    }

                    //判断bindingKey是否合法
                    if (!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] bindingKey不合法，bindingKey = " + bindingKey);
                    }

                    //判断绑定是否存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null){
                        throw new MqException("[VirtualHost] 绑定已存在， exchangeName = " + exchangeName +
                                ",queueName = " + queueName);
                    }
                    //创建绑定
                    Binding binding = new Binding();
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    binding.setBindingKey(bindingKey);

                    //判断队列和交换机是否持久话，只有两个都持久化才将绑定写入到硬盘中
                    if (existsExchange.isDurable() && existsQueue.isDurable()){
                        //写入硬盘
                        diskDataCenter.insertBinding(binding);
                    }
                    //写入内存
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 成功创建绑定，queueName = " + queueName +
                            ", exchangeName = " + exchangeName);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定创建失败，queueName = " + queueName +
                    ",exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //删除绑定
    public boolean queueUnBind(String exchangeName,String queueName){
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;

        try {
            synchronized (exchangeLocker) {
                synchronized (queueLock) {
                    //查看绑定是否存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    //不存在则抛出异常
                    if (existsBinding == null) {
                        throw new MqException("[VirtualHost] 删除绑定失败，绑定不存在，exchangeName = " + exchangeName +
                                ", queueName = " + queueName);
                    }
                    //存在则进行删除
                    //从硬盘中进行删除
                    diskDataCenter.deleteBinding(existsBinding);
                    //从内存中进行删除
                    memoryDataCenter.deleteBinding(existsBinding);
                }
            }
            System.out.println("[VirtualHost] 成功删除绑定，exchangeName = " + exchangeName +
                    ", queueName = " + queueName);
            return true;
        } catch (Exception e){
            System.out.println("[VirtualHost] 删除绑定失败，exchangeName = " + exchangeName + ", queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到对应的队列和交换机中
    public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body){
        try{
            exchangeName = virtualHostName + exchangeName;

            //验证routingKey是否合法
            if (!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] 发送消息失败，routingKey不合法，routingKey = " + routingKey);
            }
            //获取交换机
            Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
            if (existsExchange == null){
                //如果交换机不存在
                throw new MqException("[VirtualHost] 发送消息失败，交换机不存在，exchangeName = " + exchangeName);
            }


            //根据交换机的类型来进行不同方式的发送消息
            if (existsExchange.getType() == ExchangeType.DIRECT){
                //按照直接交换机的方式来转发消息
                //以routingKey作为队列名，直接把消息写入指定的队列中
                //此时，可以无视绑定关系
                String queueName = virtualHostName + routingKey;
                //构造消息对象
                Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                //获取队列
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue == null){
                    //如果队列不存在
                    throw new MqException("[VirtualHost] 发送消息失败，队列不存在，queueName = " + queueName);
                }
                //队列存在则直接发信息给队列
                sendMessage(existsQueue,message);
            }else {
                //获取所有的绑定
                ConcurrentHashMap<String, Binding> bindings = memoryDataCenter.getBindings(exchangeName);
                //遍历所有的绑定，获取队列的名字
                for (Map.Entry<String,Binding> entry : bindings.entrySet()){
                    Binding binding = entry.getValue();
                    String bindQueueName = binding.getQueueName();
                    //判断对应的队列是否存在
                    MSGQueue queue = memoryDataCenter.getQueue(bindQueueName);
                    if (queue == null){
                        //不存在则打印消息
                        System.out.println("[VirtualHost] 发送消息失败，队列不存在，queueName = " + bindQueueName);
                        continue;
                    }
                    //构造消息
                    Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                    //判断消息应该发给哪个队列
                    if (!router.rout(existsExchange.getType(), binding, message)){
                        continue;
                    }
                    //发送消息
                    sendMessage(queue,message);
                }
            }
            System.out.println("[VirtualHost] 成功发送消息，exchangeName = " + exchangeName);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 发送消息失败，exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }

    }

    //具体发送消息的逻辑
    public void sendMessage(MSGQueue queue,Message message) throws IOException, MqException, InterruptedException {
        //判断该消息是否需要持久化
        int deliverMode = message.getDeliverMode();
        //deliverMode 为1 不持久化， 为2 持久化
        //如果持久化的话存进硬盘
        if (deliverMode == 2){
            //写入硬盘、
            diskDataCenter.sendMessage(queue,message);
        }
        //存入内存
        memoryDataCenter.senMessage(queue,message);

        //通知消费者可以消费消息了
        consumerManage.notifyConsume(queue.getName());
    }

    //订阅消息
    //添加一个队列的订阅者，当队列收到消息后，就要把消息推送给相应的订阅者
    //consumerTag： 消费者的身份标识
    //autoAck： 消费者消费完消息后的应答方式，true-自动应答，false-手动应答
    //consumer：是一个回调函数，此处设定为一个函数式接口，这样后续调用 basicConsumer 的时候，就可以写作 lambda 的样子
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        //创建一个ConsumerEev对象，并找到对应的队列，将该对象添加到对应的队列去
        queueName = virtualHostName + queueName;
        try {
            //具体的业务逻辑放入到 ConsumerManger 类中去进行实现
            consumerManage.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] 成功添加订阅者，consumerTag = " + consumerTag + ", queueName = " + queueName);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 添加订阅者失败，consumerTag = " + consumerTag + ", queueName = " + queueName);
            e.printStackTrace();
            return false;
        }

    }

    //手动返回ack
    public boolean basicAck(String queueName,String messageId) throws MqException, IOException, ClassNotFoundException {
        queueName = virtualHostName + queueName;
        try {
            //获取队列和消息
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null){
                //队列不存在
                throw new MqException("[VirtualHost] 确认消息失败，队列不存在，queueName = " + queueName);
            }
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null){
                //消息不存在
                throw new MqException("[VirtualHost] 确认消息失败，消息不存在,messageId = " + messageId);
            }
            //将要删除的消息加入到待确认的集合中
            memoryDataCenter.addQueueMessageWaitAckMap(queueName,message);
            //进行删除
            //删除硬盘上的消息
            if (message.getDeliverMode() == 2){
                diskDataCenter.deleteMessage(queue,message);
            }
            //删除内存上的消息
            memoryDataCenter.deleteMessage(messageId);
            //删除待确认集合中的消息
            memoryDataCenter.deleteQueueMessageWaitAckMap(queueName,messageId);
            System.out.println("[VirtualHost] basicAck 成功，消息被确认成功！queueName = " + queueName
                 + ", messageId = " + messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicAck 成功，消息被确认失败！queueName = " + queueName
                    + ", messageId = " + messageId);
            e.printStackTrace();
            return false;
        }
    }


}